package com.jg.visitor.kafka;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.ListTopicsResult;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.common.KafkaFuture;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.List;
import java.util.Properties;
import java.util.Set;
import java.util.stream.Collectors;

@Slf4j
@Component
public class KafkaTopicCreator {
    @Value("${spring.kafka.bootstrap-servers}")
    private String serverAddress;
    @Value("${spring.kafka.consume-topics}")
    private List<String> topics;

    @PostConstruct
    public void checkAndCreateTopic() {
        try{
            Properties prop = new Properties();
            prop.put("bootstrap.servers",serverAddress);

            AdminClient adminClient = AdminClient.create(prop);

            ListTopicsResult result = adminClient.listTopics();

            KafkaFuture<Set<String>> names = result.names();

            Set<String> exitsTopics = names.get();
            log.info("exitsTopics={}",exitsTopics);

            List<NewTopic> newTopics = topics
                    .stream()
                    .filter(topic -> !exitsTopics.contains(topic))
                    .map(topic -> new NewTopic(topic, 1, (short) 1))
                    .collect(Collectors.toList());
            log.info("newTopics={}",newTopics);

            adminClient.createTopics(newTopics);
        }catch (Exception e){
            log.error("checkAndCreateTopic error",e);
        }
    }
}
